package com.gaogzhen.designPattern.producerConsumer.workSteal;

import com.gaogzhen.designPattern.twoPhraseTermination.AbstractTerminatableThread;
import com.gaogzhen.designPattern.twoPhraseTermination.TerminationToken;

import java.util.Random;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 * @version 1.0
 * @description 生产者
 * @date 2022-10-28 10:38
 */
public class WorkStealExample{
    private final WorkStealingEnabledChannel<String> channel;
    private final TerminationToken token = new TerminationToken();

    public WorkStealExample() {
        int nCpu = Runtime.getRuntime().availableProcessors();
        int consumerCount = nCpu / 2 + 1;
        BlockingDeque<String>[] managedQueues = new LinkedBlockingDeque[consumerCount];
        // 通道绑定多个队列
        channel = new WorkStealingChannel<String>(managedQueues);
        Consumer[] consumers = new Consumer[consumerCount];
        for (int i = 0; i < consumerCount; i++) {
            managedQueues[i] = new LinkedBlockingDeque<String>();
            consumers[i] = new Consumer(token, managedQueues[i]);
        }
        for (int i = 0; i < nCpu; i++) {
            new Producer().start();
        }

        for (int i = 0; i < consumerCount; i++) {
            consumers[i].start();
        }
    }

    public void doSomething() {

    }

    public static void main(String[] args) throws InterruptedException {
        WorkStealExample wse = new WorkStealExample();
        wse.doSomething();
        TimeUnit.MILLISECONDS.sleep(2000);
    }

    private class Producer extends AbstractTerminatableThread {
        private int i = 0;

        @Override
        protected void doRun() throws Exception {
            channel.put(String.valueOf(i++));
            token.reservations.incrementAndGet();
        }
    }

    private class Consumer extends AbstractTerminatableThread{

        private final BlockingDeque<String> workQueue;
        private final Random r  = new Random();

        public Consumer(TerminationToken token, BlockingDeque<String> workQueue) {
            super(token);
            this.workQueue = workQueue;
        }

        @Override
        protected void doRun() throws Exception {
            String product = channel.take(workQueue);
            System.out.println("Processing product:" + product);
            // 模拟耗时
            try {
                TimeUnit.MILLISECONDS.sleep(r.nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                token.reservations.decrementAndGet();
            }

        }
    }
}
